package com.example.socket.client;

import java.util.concurrent.*;

/**
 * Socket异步请求回应Future
 * @param <T>
 */
public class SocketFuture<T> implements Future<T> {
	
	public static final int TIME_OUT = 5000;

	private final SocketCallback<T> callback;

	private volatile boolean completed;
	private volatile boolean cancelled;
	private volatile T result;
	private volatile Exception ex;

	private final int serialNumber;
	@SuppressWarnings("rawtypes")
	private final ConcurrentMap<Integer, SocketFuture> futures;
	private Client client;
	private int timeOut = TIME_OUT;

	@SuppressWarnings("rawtypes")
	public SocketFuture(ConcurrentMap<Integer, SocketFuture> futures, int sn, final SocketCallback<T> callback, Client client, int timeOut) {
		this.callback = callback;
		this.serialNumber = sn;
		this.futures = futures;
		this.client = client;
		this.timeOut = timeOut;
	}

	@Override
	public boolean isCancelled() {
		return this.cancelled;
	}

	@Override
	public boolean isDone() {
		return this.completed;
	}

	private T getResult() throws ExecutionException {
		if (this.ex != null) {
			throw new ExecutionException(this.ex);
		}
		return this.result;
	}

	@Override
	public synchronized T get() throws InterruptedException, ExecutionException {
		try {
			return get(timeOut, TimeUnit.MILLISECONDS);
		} catch (TimeoutException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public synchronized T get(final long timeout, final TimeUnit unit) throws InterruptedException, ExecutionException,
			TimeoutException {
		final long msecs = unit.toMillis(timeout);
		final long startTime = (msecs <= 0) ? 0 : System.currentTimeMillis();
		long waitTime = msecs;
		if (this.completed) {
			return getResult();
		} else if (waitTime <= 0) {
			throw new TimeoutException();
		} else {
			for (;;) {
				wait(waitTime);
				if (this.completed) {
					return getResult();
				} else {
					waitTime = msecs - (System.currentTimeMillis() - startTime);
					if (waitTime <= 0) {
						throw new TimeoutException();
					}
				}
			}
		}
	}

	public boolean onSuccess(final T result) {
		// 移除回调
		futures.remove(serialNumber);
		synchronized (this) {
			if (this.completed) {
				return false;
			}
			this.completed = true;
			this.result = result;
			notifyAll();
		}
		if (this.callback != null) {
			try {
				this.callback.onSuccess(result);
			} catch (Exception e) {
				// 设置异常
				this.callback.onError(e);
			}
		}
		return true;
	}

	public boolean onError(final Exception exception) {
		// 移除回调
		futures.remove(serialNumber);
		synchronized (this) {
			if (this.completed) {
				return false;
			}
			this.completed = true;
			this.ex = exception;
			notifyAll();
		}
		if (this.callback != null) {
			this.callback.onError(exception);
		}
		return true;
	}

	@Override
	public boolean cancel(final boolean mayInterruptIfRunning) {
		// 移除回调
		futures.remove(serialNumber);
		synchronized (this) {
			if (this.completed) {
				return false;
			}
			this.completed = true;
			this.cancelled = true;
			notifyAll();
		}
		if (this.callback != null) {
			// 设置异常
			this.callback.onCancel();
		}
		return true;
	}
	
	public void timeOut(){
		// 移除回调
		futures.remove(serialNumber);
		synchronized (this) {
			if (this.completed) {
				return ;
			}
			this.completed = true;
			this.cancelled = true;
			notifyAll();
		}
		if (this.callback != null) {
			// 设置异常
			this.callback.onTimeout();
		}
	}

	public boolean cancel() {
		return cancel(true);
	}

	public Client getClient() {
		return client;
	}
}